// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/deribit_base/feed_v2/processor.h"

#include <set>
#include <string_view>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/deribit_base/feed_v2/update.h"
#include "coin2/exchange/deribit_options/symbology/product.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::deribit_base::feed_v2::impl {
void DeribitBaseFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  using deribit_options::symbology::DeribitOptionsProduct;

  upd_mgr_.onBeginPacket(packet);

  simdjson::dom::object obj;
  simdjson::error_code err;
  parser_.parse(packet.payload).get<simdjson::dom::object>().tie(obj, err);

  if (err) {
    Error("[Deribit Feed Processor] err", packet.payload);
    return;
  }

  std::string_view ch;
  std::string_view native_symbol;
  std::string_view url;
  obj["url"].get<std::string_view>().tie(url, err);
  if (err) {
    // no url, websocet channel
    obj["params"]["channel"].get<std::string_view>().tie(ch, err);
    if (err) {
      std::string_view error_message;
      obj["error"]["message"].get<std::string_view>().tie(error_message, err);
      if (!err && error_message != "bad_request") {
        LOG(INFO) << "[Deribit Feed Processor] meta msg for " << packet.topic_name << ": "
                << packet.payload.data();
      }
      return;
    }

    // book/trade channle symbol in the second field
    size_t dot1 = ch.find(".");
    size_t dot2 = ch.find(".", dot1 + 1);
    std::string ch_text(ch);
    ch_text.replace(dot1 + 1, dot2 - dot1 - 1, "{}");
    if (!CheckChannelFilter(ch_text)) {
      return;
    }
    native_symbol = ch.substr(dot1 + 1, dot2 - dot1 - 1);
  } else {
    // has url, rest channel
    std::string_view ch_kline = "/public/get_tradingview_chart_data?";
    if (url.find(ch_kline) != std::string_view::npos) {
      ch = "kline_rest";
      if (!CheckChannelFilter("kline_rest")) {
        return;
      }
    }
  }

  if (StringStartsWith(ch, "trades")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      if (dynamic_add_product_) {
        auto product = DeribitOptionsProduct::FromNativeStrNoThrow(
            std::string(native_symbol), packet.timestamp);
        if (product) {
          AddProduct(*product.get(), subscriber_);
        } else {
          LOG_FIRST_N(INFO, 100) << "[Deribit Feed Processor] unknown symbol: "
              << native_symbol << ", " << packet.payload;
          return;
        }
      } else {
        return;
      }
    }
    Trade* trade = upd_mgr_.GetTradeMutable();
    int index = 0;
    int last_index = obj["params"]["data"].get<simdjson::dom::array>().size();
    for (const auto& trade_json : obj["params"]["data"].get<simdjson::dom::array>()) {
      DeribitBaseFeedUpdateParser::ParseTrade(trade_json, trade);
      upd_mgr_.PublishTrade(native_symbol, index != last_index);
    }
  } else if (StringStartsWith(ch, "book")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      if (dynamic_add_product_) {
        auto product = DeribitOptionsProduct::FromNativeStrNoThrow(
            std::string(native_symbol), packet.timestamp);
        if (product) {
          AddProduct(*product.get(), subscriber_);
        } else {
          LOG_FIRST_N(INFO, 100) << "[Deribit Feed Processor] unknown symbol: "
              << native_symbol << ", " << packet.payload;
          return;
        }
      } else {
        return;
      }
    }
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      if (publish_bbo_book_only_) {
        std::optional<PriceQty> bid_pre = book->Bid0();
        std::optional<PriceQty> ask_pre = book->Ask0();
        if (DeribitBaseFeedUpdateParser::ParseBook(
                book,
                obj["params"]["data"].get<simdjson::dom::object>(),
                BookBuilder::LIGHT_BBO,
                native_symbol,
                packet.timestamp)) {
          std::optional<PriceQty> bid_post = book->Bid0();
          std::optional<PriceQty> ask_post = book->Ask0();
          bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
          if (is_bbo_changed) {
            upd_mgr_.PublishBook();
          }
        }
      } else {
        DeribitBaseFeedUpdateParser::ParseBook(
            book,
            obj["params"]["data"].get<simdjson::dom::object>(),
            BookBuilder::DELTA,
            native_symbol,
            packet.timestamp);
        upd_mgr_.PublishBook();
      }
    }
  } else if (StringStartsWith(ch, "ticker")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      if (dynamic_add_product_) {
        auto product = DeribitOptionsProduct::FromNativeStrNoThrow(
            std::string(native_symbol), packet.timestamp);
        if (product) {
          AddProduct(*product.get(), subscriber_);
        } else {
          LOG_FIRST_N(INFO, 100) << "[Deribit Feed Processor] unknown symbol: "
              << native_symbol << ", " << packet.payload;
          return;
        }
      } else {
        return;
      }
    }
    const auto& msg = obj["params"]["data"];
    auto* index = upd_mgr_.GetIndexMutable();
    DeribitBaseFeedUpdateParser::ParseIndex(msg, index);
    upd_mgr_.PublishIndexWithExchangeTimeCheck(std::string(native_symbol));

    auto* mark_price = upd_mgr_.GetMarkPriceMutable();
    DeribitBaseFeedUpdateParser::ParseMarkPrice(msg, mark_price);
    upd_mgr_.PublishMarkPriceWithExchangeTimeCheck(std::string(native_symbol));

    auto* open_interest = upd_mgr_.GetOpenInterestMutable();
    DeribitBaseFeedUpdateParser::ParseOpenInterest(msg, open_interest);
    upd_mgr_.PublishOpenInterestWithExchangeTimeCheck(std::string(native_symbol));

    auto* funding_rate = upd_mgr_.GetFundingRateMutable();
    if (DeribitBaseFeedUpdateParser::ParseFundingRate(msg, funding_rate)) {
      upd_mgr_.PublishFundingRateWithExchangeTimeCheck(std::string(native_symbol));
    }

    // Only available for options case.
    if (dynamic_add_product_) {
      auto* iv_greeks = upd_mgr_.GetIvGreeksMutable();
      if (DeribitBaseFeedUpdateParser::ParseIvGreeks(msg, iv_greeks)) {
        upd_mgr_.PublishIvGreeksWithExchangeTimeCheck(std::string(native_symbol));
      }
    }
  } else if (ch == "kline_rest") {
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    native_symbol = parameters["instrument_name"];
    auto& interval = parameters["resolution"];
    fastfeed::proto::KlineInterval kinterval =
        DeribitBaseFeedUpdateParser::ParseKlineInterval(interval);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, kinterval);
    if (klines != nullptr) {
      auto ticks = obj["data"]["result"]["ticks"].get<simdjson::dom::array>();
      std::optional<int64_t> first_open_timestamp;
      if (ticks.size() > 0) {
        first_open_timestamp = ticks.at(0).get<int64_t>() * 1'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      if (DeribitBaseFeedUpdateParser::ParseKlines(
          obj, klines, packet.timestamp, kinterval,
          upd_mgr_.mea().market == MarketType::Futures)){
         upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else if (StringStartsWith(ch, "deribit_price_index")) {
    if (upd_mgr_.is_symbol_registered(native_symbol)) {
      const auto& msg = obj["params"]["data"];
      auto* index = upd_mgr_.GetIndexMutable();
      DeribitBaseFeedUpdateParser::ParseIndex(msg, index);
      upd_mgr_.PublishIndexWithExchangeTimeCheck(std::string(native_symbol));
    }
  } else {
    Error("[Deribit Feed Processor] unknown channel: " + std::string(ch), packet.payload);
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::deribit_base::feed_v2::impl
